/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.rtd.engine.main;

import com.semidata.rtd.core.service.ServiceManager;
import com.semidata.rtd.engine.executor.TaskExecutor;
import com.semidata.rtd.engine.job.AbstractJob;
import com.semidata.rtd.engine.job.JobCompleteHandler;
import com.semidata.rtd.engine.job.Task;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class TestJob extends AbstractJob {

	private static Monitor monitor = ServiceManager.getService(Monitor.class);
	private long task1StartTime = 0;
	private long task2StartTime = 0;
	private long task3StartTime = 0;

	public TestJob(TaskExecutor executor) {
		super(executor, new JobCompleteHandler() {
			private long startTime = System.currentTimeMillis();

			@Override
			public void handle() {
				long time = System.currentTimeMillis() - startTime;
				monitor.totalJobCount.incrementAndGet();
				monitor.totalJobTime.addAndGet(time);
			}
		}, new JobCompleteHandler() {
			private long startTime = System.currentTimeMillis();

			@Override
			public void handle() {
				long time = System.currentTimeMillis() - startTime;
				monitor.totalJobCount.incrementAndGet();
				monitor.totalJobTime.addAndGet(time);
				monitor.errorJobCount.incrementAndGet();
				monitor.errorJobTime.addAndGet(time);

			}
		});

	}

	@Override
	protected void schedule(Task task) {
		
		if (task == null) {
			task1StartTime = System.currentTimeMillis();
			getExecutor().submit(new TestTask1(this));
		} else if (task.getClass() == TestTask1.class) {
			long task1Time = System.currentTimeMillis() - task1StartTime;
			
			monitor.totalTaskCount.incrementAndGet();
			monitor.totalTaskTime.addAndGet(task1Time);
			monitor.totalTask1Count.incrementAndGet();
			monitor.totalTask1Time.addAndGet(task1Time);
			
			Exception exception = task.getException();
			if (exception != null) {
				exception.printStackTrace();
				monitor.errorTaskCount.incrementAndGet();
				monitor.errorTaskTime.addAndGet(task1Time);
				monitor.errorTask1Count.incrementAndGet();
				monitor.errorTask1Time.addAndGet(task1Time);
				error();
			} else {
				task2StartTime = System.currentTimeMillis();
				getExecutor().submit(new TestTask2(this));
			}
		} else if (task.getClass() == TestTask2.class) {
			long task2Time = System.currentTimeMillis() - task2StartTime;
			monitor.totalTaskCount.incrementAndGet();
			monitor.totalTaskTime.addAndGet(task2Time);
			monitor.totalTask2Count.incrementAndGet();
			monitor.totalTask2Time.addAndGet(task2Time);
			Exception exception = task.getException();
			if (exception != null) {
				monitor.errorTaskCount.incrementAndGet();
				monitor.errorTaskTime.addAndGet(task2Time);
				monitor.errorTask2Count.incrementAndGet();
				monitor.errorTask2Time.addAndGet(task2Time);
			}
			task3StartTime = System.currentTimeMillis();
			getExecutor().submit(new TestTask3(this, 10));
		} else if (task.getClass() == TestTask3.class) {
			long task3Time = System.currentTimeMillis() - task3StartTime;
			
			monitor.totalTaskCount.incrementAndGet();
			monitor.totalTaskTime.addAndGet(task3Time);
			monitor.totalTask3Count.incrementAndGet();
			monitor.totalTask3Time.addAndGet(task3Time);
			
			Exception exception = task.getException();
			if (exception != null) {
				monitor.errorTaskCount.incrementAndGet();
				monitor.errorTaskTime.addAndGet(task3Time);
				monitor.errorTask3Count.incrementAndGet();
				monitor.errorTask3Time.addAndGet(task3Time);
				error();
			} else {
				done();
			}
		}

	}

}
